package com.mogician.rabbitmq_demo.service;

import android.util.Log;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.mogician.rabbitmq_demo.bean.MyMessage;
import com.mogician.rabbitmq_demo.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.util.concurrent.BlockingDeque;
import java.util.concurrent.LinkedBlockingDeque;

public class MQPublishService
{
    // 创建内部消息队列，供生产者发布消息用
    private BlockingDeque<MyMessage> queue;
    private ConnectionFactory factory;
    private ObjectMapper mapper;
    private Connection connection;
    private Channel ch;

    public MQPublishService(final String exchangeName, final String routingKey)
    {
        queue = new LinkedBlockingDeque<>();
        factory = ConnectionUtil.setUpConnectionFactory();
        mapper = new ObjectMapper();
        publishToAMPQ(exchangeName, routingKey);
    }

    private void publishToAMPQ(final String exchangeName, final String routingKey)
    {
        Thread publishThread = new Thread(new Runnable()
        {
            @Override
            public void run()
            {
                while (true)
                {
                    try
                    {
//                        // 创建连接
//                        Connection connection = factory.newConnection();
//                        // 创建通道
//                        Channel ch = connection.createChannel();
                        if(connection == null)
                            connection = factory.newConnection();
                        if(ch == null)
                            ch = connection.createChannel();
                        ch.confirmSelect();
                        while (true)
                        {
                            MyMessage message = queue.takeFirst();
                            try
                            {
                                // 发布消息
                                ch.basicPublish(exchangeName, routingKey,
                                        null, mapper.writeValueAsBytes(message));
                                ch.waitForConfirmsOrDie();

                            } catch (Exception e)
                            {
                                queue.putFirst(message);
                                throw e;
                            }

                        }
                    } catch (InterruptedException e)
                    {
                        break;
                    } catch (Exception e)
                    {
                        Log.d("TAG_Publish", "Connection broken: " + e.getClass().getName());
                        try
                        {
                            Thread.sleep(5000); //sleep and then try again
                        } catch (InterruptedException e1)
                        {
                            break;

                        }
                    }
                }
            }
        });

        publishThread.start();

    }

    public void publishMessage(MyMessage message) {
        //向内部阻塞队列添加一条消息
        try {
            queue.putLast(message);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}